﻿namespace NTLib.Pipe.Streaming
{
    using Containers;
    using Descriptors;
    using Pipe.Containers;
    using Pipe.Descriptors;
    using System;
    using System.Collections.Generic;
    using System.Linq;
    using System.Text;
    using System.Threading.Tasks;
    using System.Threading;
    using Exceptions;
    using Formats;
    using Core.Exceptions;
    using Resources;
    using Core.Extensions;
    using Core.Manipulators;
    using Core.Threading;
    using System.Runtime.CompilerServices;
    using Pipe.Exceptions;
    using Core.Log;
    using Descriptors.Packages;
    using System.IO;
    using Core.Args;
    using Core.Timers;
    using System.Diagnostics;

    /// <summary>
    /// Base pipe source that managed all the information 
    /// using serialization, streaming and deserialization
    /// </summary>
    /// <typeparam name="TType">The type of the type.</typeparam>
    /// <typeparam name="TTargetDescriptor">The type of the target descriptor.</typeparam>
    /// <seealso cref="NTLib.Pipe.Containers.BasePipeSourcePoolItem{TType, TTargetDescriptor}" />
    public abstract class BaseStreamingPipeSourcePoolItem<TType, TTargetDescriptor>
        : BasePipeSourcePoolItem<TType, TTargetDescriptor>
        where TTargetDescriptor : IStreamingTargetDescriptor

    {
        #region Fields

        private readonly Queue<IPackage> _read;
        private readonly SemaphoreLocker _locker;
        private readonly ReaderWriterLocker _resourceAccess;
        private readonly Queue<TaskCompletionSource<object>> _readWaiter;

        private Timer _aliveTimer;

        private readonly Queue<IPackage> _toSend;
        //private readonly Dictionary<Guid, Tuple<Type, TaskCompletionSource<object>>> _answers;

        private CancellationTokenSource _workingToken;
        private IPackageFormater _formater;

        #endregion

        #region Ctor

        /// <summary>
        /// Initializes a new instance of the <see cref="BaseStreamingPipeSourcePoolItem{TType, TTargetDescriptor}"/> class.
        /// </summary>
        public BaseStreamingPipeSourcePoolItem()
        {
            this._read = new Queue<IPackage>();
            this._locker = new SemaphoreLocker();
            this._readWaiter = new Queue<TaskCompletionSource<object>>();
            this._resourceAccess = new ReaderWriterLocker();

            this._toSend = new Queue<IPackage>();
            //this._answers = new Dictionary<Guid, Tuple<Type, TaskCompletionSource<object>>>();
        }

        #endregion

        #region Properties

        /// <summary>
        /// Gets a value indicating whether this instance can pop item.
        /// </summary>
        /// <value>
        /// <c>true</c> if this instance can pop item; otherwise, <c>false</c>.
        /// </value>
        public override bool CanPopItem
        {
            get
            {
                base.CheckSourceIsInitialized();
                return this.PipeType == PipeType.FromTarget ||
                       this.PipeType == PipeType.TwoWay;
            }
        }

        /// <summary>
        /// Gets the value indicating that the current pipe allow the user to push items through it.
        /// </summary>
        /// <value>
        /// s
        /// <c>true</c> if this instance can push item; otherwise, <c>false</c>.
        /// </value>
        public override bool CanPushItem
        {
            get
            {
                base.CheckSourceIsInitialized();

                return this.PipeType == PipeType.OneWay ||
                       this.PipeType == PipeType.TwoWay;
            }
        }

        #endregion

        #region Methods

        /// <summary>
        /// Initializes the current pipe asynchronously.
        /// </summary>
        /// <param name="token"></param>
        /// <returns>
        /// Task used to make this initialization asynchronously.
        /// </returns>
        protected override async Task OnInitializeAsync(CancellationToken token = default(CancellationToken))
        {
            await base.OnInitializeAsync(token);

            this._formater = null;

            if (base.TargetTyped == null ||
                base.TargetTyped.PackageFormater == null)
            {
                throw new MissingFormaterException();
            }

            token.ThrowIfCancellationRequested();

            if (PackageFormatManager.Instance == null)
            {
                throw new DependencyMissingException(typeof(PackageFormatManager));
            }

            token.ThrowIfCancellationRequested();

            var formater = PackageFormatManager.Instance.Create(base.TargetTyped.PackageFormater.Id);

            token.ThrowIfCancellationRequested();

            if (formater == null)
            {
                throw new MissingFormaterException(InternErrorMessages.FormaterNotFound.Argument(base.TargetTyped.PackageFormater.Id));
            }

            token.ThrowIfCancellationRequested();

            this._formater = formater;
            this._workingToken = new CancellationTokenSource();

            token.ThrowIfCancellationRequested();

            await OnInitializeAfterCheckAsync(token);

            token.ThrowIfCancellationRequested();

            if (this.PipeType == PipeType.TwoWay ||
                this.PipeType == PipeType.FromTarget)
            {
                StreamReadingLoop();
            }
        }

        /// <summary>
        /// Called when the update pop waiter is called.
        /// This methods is used to dequeue item from the queue to the waiter.
        /// This calls is Thread Safe.
        /// </summary>
        /// <param name="token"></param>
        /// <returns></returns>
        protected override async Task<TType> OnDequeueItemAsync(CancellationToken token)
        {
            CheckInitialized();

            TaskCompletionSource<object> _order = null;
            IPackage pkg = null;
            using (await this._locker.ScopeLockAsync(false, token))
            {
                if (this._read.Count > 0 && this._readWaiter.Count == 0)
                    pkg = this._read.Dequeue();
                else
                {
                    _order = new TaskCompletionSource<object>();
                    this._readWaiter.Enqueue(_order);
                }
            }

            if (pkg == null && _order != null)
            {
                await _order.Task;
                pkg = (IPackage)_order.Task.Result;
            }

            using (this._resourceAccess.ScopeLock(true))
            {
                return this._formater.FromPackage<TType>(pkg);
            }
        }

        ///// <summary>
        ///// Pushes the item asynchronous on the storage system.
        ///// </summary>
        ///// <typeparam name="U">Type of the data wait after the push</typeparam>
        ///// <param name="item">The item to push.</param>
        ///// <param name="waiter">The task completion source used to wait a possible answer.</param>
        ///// <param name="token">The token to cancel the current task.</param>
        ///// <returns>
        ///// Task to wait when the push it's finish.
        ///// </returns>
        //protected override async Task OnPushItemAsync<U>(TType item, TaskCompletionSource<object> waiter, CancellationToken token)
        //{
        //    CheckInitialized();
        //    using (await this._locker.ScopeLockAsync(false, token))
        //    {
        //        IPackage pkg = this._formater.ToPackage<TType>(item);
        //        this._toSend.Enqueue(pkg);
        //        this._answers.Add(pkg.Id, new Tuple<Type, TaskCompletionSource<object>>(typeof(U), waiter));
        //    }
        //    StreamSendLoop();
        //}

        /// <summary>
        /// Pushes the item on the storage system without waiting any response.
        /// </summary>
        /// <param name="item"></param>
        /// <param name="token"></param>
        protected override async Task OnSend(TType item, CancellationToken token)
        {
            CheckInitialized();
            using (await this._locker.ScopeLockAsync(false, token))
            {
                this._toSend.Enqueue(this._formater.ToPackage<TType>(item));
            }
            StreamSendLoop();
        }

        /// <summary>
        /// Checks if the instance is initialized.
        /// </summary>
        [MethodImpl(MethodImplOptions.AggressiveInlining)]
        protected virtual void CheckInitialized()
        {
            if (this._formater == null)
                throw new PipeNotInitializedException();
        }
        /// <summary>
        /// Called when the pool item is reset.
        /// Everything MUST be initialized on this method.
        /// </summary>
        protected override void OnPoolItemReset()
        {
            CleanContent();

            base.OnPoolItemReset();
        }

        /// <summary>
        /// Manageds the dispose.
        /// </summary>
        protected override void ManagedDispose()
        {
            CleanContent();

            base.ManagedDispose();
        }

        /// <summary>
        /// Cleans the content.
        /// </summary>
        private void CleanContent()
        {
            if (this._workingToken != null && !this._workingToken.IsCancellationRequested)
            {
                var tmp = this._workingToken;
                this._workingToken = null;
                tmp.Cancel();
                tmp.Dispose();
            }

            this._formater = null;

            using (this._locker.ScopeLock())
            {
                this._read.Clear();
                this._toSend.Clear();

                var localCpy = this._readWaiter.ToArray();
                this._readWaiter.Clear();

                foreach (var local in localCpy)
                {
                    try
                    {
                        local.TrySetCanceled();
                    }
                    catch (Exception ex)
                    {
                        NTLog.Instance.RecordError(ex, this.GetType().Name);
                    }
                }

                //var answerCopy = this._answers.ToArray();
                //this._answers.Clear();
                //foreach (var local in answerCopy)
                //{
                //    try
                //    {
                //        local.Value.Item2.TrySetCanceled();
                //    }
                //    catch (Exception ex)
                //    {
                //        NTLog.Instance.RecordError(ex, this.GetType().Name);
                //    }
                //}

                try
                {
                    OnCleanContent();
                }
                catch
                {

                }
            }
        }

        /// <summary>
        /// Called when the pipe is cleaned
        /// </summary>
        protected abstract void OnCleanContent();

        /// <summary>
        /// Loop that will read the stream
        /// </summary>
        private async void StreamReadingLoop()
        {
            var localWorkingToken = this._workingToken;

            try
            {
                IPackageFormater formatter;
                do
                {
                    using (this._resourceAccess.ScopeLock(true))
                    {
                        formatter = this._formater;
                    }

                    if (localWorkingToken != null &&
                        !localWorkingToken.IsCancellationRequested &&
                        formatter != null)
                    {
                        localWorkingToken.Token.ThrowIfCancellationRequested();

                        if (await CanReadPackageAsync(localWorkingToken.Token))
                        {
                            try
                            {
                                var packageRead = await ReadAsync(this._formater.ReaderDescriptor, localWorkingToken.Token);
                                if (packageRead != null)
                                {
                                    using (await this._locker.ScopeLockAsync(false, localWorkingToken.Token))
                                    {
                                        localWorkingToken.Token.ThrowIfCancellationRequested();

                                        Tuple<Type, TaskCompletionSource<object>> waiter;
                                        //if (this._answers.TryGetValue(packageRead.Id, out waiter))
                                        //{
                                        //    waiter.Item2.TrySetResult(formatter.FromPackage(packageRead, waiter.Item1));
                                        //    this._answers.Remove(packageRead.Id);
                                        //}
                                        //else
                                        //{
                                        if (this._readWaiter.Count == 0)
                                        {
                                            this._read.Enqueue(packageRead);
                                        }
                                        else
                                        {
                                            var readWaiter = this._readWaiter.Dequeue();
                                            readWaiter.TrySetResult(packageRead);
                                            continue;
                                        }
                                        //}
                                    }

                                    localWorkingToken.Token.ThrowIfCancellationRequested();
                                    base.RaiseDataReceived();
                                }
                            }
                            catch (InvalidDataException)
                            {
                                if (localWorkingToken.IsCancellationRequested ||
                                    !await this.CanReadPackageAsync(localWorkingToken.Token))
                                {
                                    this.CleanContent();
                                    return;
                                }
                            }
                            catch (PipeException)
                            {

                            }
                        }
                        else
                        {
                            await Task.Delay(1);
                        }
                    }
                } while (localWorkingToken != null && !localWorkingToken.IsCancellationRequested);
            }
            catch (OperationCanceledException)
            {
            }
        }

        /// <summary>
        /// Raises the data received.
        /// </summary>
        protected override void RaiseDataReceived()
        {
            CancellationTokenSource source = null;
            using (this._resourceAccess.ScopeLock(true))
            {
                source = this._workingToken;
            }
            var lockerTask = this._locker.ScopeLockAsync(false, source.Token);
            lockerTask.Wait(source.Token);
            using (lockerTask.Result)
            {
                if (this._readWaiter.Count > 0)
                {
                    var readWaiter = this._readWaiter.Dequeue();
                    readWaiter.SetResult(42);
                    return;
                }
            }
            base.RaiseDataReceived();
        }

        /// <summary>
        /// Loop that will send into the stream
        /// </summary>
        private async void StreamSendLoop()
        {
            var localWorkingToken = this._workingToken;

            IPackage pkgToSend = null;

            do
            {
                if (localWorkingToken != null && !localWorkingToken.IsCancellationRequested)
                {
                    using (await this._locker.ScopeLockAsync())
                    {
                        if (this._toSend.Count > 0)
                            pkgToSend = this._toSend.Peek();
                        else
                            pkgToSend = null;
                    }

                    if (pkgToSend != null)
                    {
                        localWorkingToken.Token.ThrowIfCancellationRequested();

                        if (await CanSendPackageAsync(pkgToSend, localWorkingToken.Token))
                        {

                            localWorkingToken.Token.ThrowIfCancellationRequested();

                            try
                            {
                                await SendPackageAsync(pkgToSend, localWorkingToken.Token);
                            }
                            catch (InvalidDataException)
                            {
                                if (localWorkingToken.IsCancellationRequested ||
                                    !await this.CanReadPackageAsync(localWorkingToken.Token))
                                {
                                    this.CleanContent();
                                    return;
                                }
                            }

                            this._toSend.Dequeue();
                            pkgToSend.Dispose();
                        }
                        else
                        {
                            await Task.Delay(1);
                        }
                    }

                    localWorkingToken.Token.ThrowIfCancellationRequested();
                }
            } while (pkgToSend != null);
        }

        protected override void OnStatusChanged(ValueChangedEventArgs<PipeStatus> args)
        {
            if (args.NewValue == PipeStatus.Ready)
            {
                this._aliveTimer = new Timer(TimeSpan.FromSeconds(0.5));
                this._aliveTimer.Elapsed += AliveTimer_Elapsed;
                this._aliveTimer.Start(this._workingToken.Token);
            }
            else if (args.OldValue == PipeStatus.Ready)
            {
                ClosePipeSource();
            }

            base.OnStatusChanged(args);
        }

        /// <summary>
        /// Tick used to check if the pipe source is alive
        /// </summary>
        private void AliveTimer_Elapsed(object obj)
        {
            Debug.WriteLine("Pipe {0} is open {1}", this.Id, IsPipeSourceAlive());
            if (!IsPipeSourceAlive())
            {
                ClosePipeSource();
            }
        }

        /// <summary>
        /// Closes the pipe source.
        /// </summary>
        protected void ClosePipeSource()
        {
            using (this._locker.ScopeLock())
            {
                if (this._aliveTimer != null)
                {
                    this._aliveTimer.Dispose();
                    this._aliveTimer = null;
                }
            }

            CleanContent();
        }

        #region Api

        /// <summary>
        /// Initializes the current pipe asynchronously.
        /// THis method is called after the formater check
        /// </summary>
        /// <param name="token">The token.</param>
        /// <returns>
        /// Task used to make this initialization asynchronously.
        /// </returns>
        protected abstract Task OnInitializeAfterCheckAsync(CancellationToken token);

        /// <summary>
        /// Determines whether this instance can read package asynchronous
        /// </summary>
        /// <param name="item">The item.</param>
        /// <param name="token">The token.</param>
        /// <remarks>
        ///     Multiple called can occured simultianiously
        /// </remarks>
        protected abstract Task<bool> CanReadPackageAsync(CancellationToken token);

        /// <summary>
        /// Reads the new item using the package formater.
        /// </summary>
        /// <param name="formater">The formater.</param>
        /// <param name="token">The token.</param>
        [ThreadSafe]
        protected abstract Task<IPackage> ReadAsync(IPackageDescriptor descriptor, CancellationToken token);

        /// <summary>
        /// Determines whether this instance can send the package.
        /// </summary>
        /// <param name="item">The item.</param>
        /// <param name="token">The token.</param>
        /// <returns>
        ///   <c>true</c> if this instance can send the package the specified item; otherwise, <c>false</c>.
        /// </returns>
        /// <remarks>
        ///     Multiple called can occured simultianiously
        /// </remarks>
        protected abstract Task<bool> CanSendPackageAsync(IPackage item, CancellationToken token);

        /// <summary>
        /// Sends the package asynchronous.
        /// </summary>
        /// <param name="package">The package.</param>
        /// <param name="token">The token.</param>
        /// <remarks>
        ///     Multiple called can occured simultianiously
        /// </remarks>
        protected abstract Task SendPackageAsync(IPackage package, CancellationToken token);

        /// <summary>
        /// Determines whether if the pipe source is alive.
        /// </summary>
        /// <returns>
        ///   <c>true</c> if the pipe source is alive; otherwise, <c>false</c>.
        /// </returns>
        protected abstract bool IsPipeSourceAlive();

        #endregion

        #endregion
    }
}
